package rabbitmq;

import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.io.UnsupportedEncodingException;

/**
 * 消息消费者
 */
public class Customer {

	/**
	 * 对应Producer.producerTest
	 */
	@Test
	public void consumerTest() {
		Connection connection = null;
		try {
			ConnectionService conn = new ConnectionService();
			connection = conn.initConn();
			Channel channel = connection.createChannel();

			//声明路由名字和类型
			channel.exchangeDeclare("test_exchange", "fanout", true, false, null);
			//声明队列
			channel.queueDeclare("firstQueue", true, false, false, null);
			//绑定路由和队列、key
			channel.queueBind("firstQueue", "test_exchange", "test_routing_key", null);

			channel.basicQos(1);

			//订阅方式
			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
					try {
						String message = "";
						try {
							message = new String(body, "UTF-8");
						} catch (UnsupportedEncodingException e) {
							e.printStackTrace();
						} catch (Throwable ex) {
							ex.printStackTrace();
						}
						System.out.println(" Consumer have received '" + message + "'");
						//应答删除
						channel.basicAck(envelope.getDeliveryTag(), false);
					} catch (Exception ex) {
						System.out.println(ex);
						try {
							channel.basicNack(envelope.getDeliveryTag(),false,true);
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
			};
			//关闭自动应答
			channel.basicConsume("firstQueue", false, consumer);
		} catch(Exception ex) {
			System.out.println(ex.getMessage()+";"+ex);
		} finally {

		}
	}


	@Test
	public void consumerTest2() {
		/**
		 * 单元测试不支持多线程，测试订阅模式-消费者会出现信息无法打印的情况
		 */
	}

	public static void main(String[] args) {
		Connection connection = null;
		try {
			ConnectionService conn = new ConnectionService();
			connection = conn.initConn();
			Channel channel = connection.createChannel();

			//声明路由名字和类型
			channel.exchangeDeclare("test_exchange", "fanout", true, false, null);
			//声明队列
			channel.queueDeclare("firstQueue", true, false, false, null);
			//绑定路由和队列，可以不设置routing_key
			channel.queueBind("firstQueue", "test_exchange", "", null);

			channel.basicQos(1);

			//订阅方式
			channel.basicConsume("firstQueue", false, new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
					try {
						String message = new String(body, "UTF-8");
						System.out.println(" Consumer have received '" + message + "'");
						Thread.sleep(200);
						//应答删除
						channel.basicAck(envelope.getDeliveryTag(), false);
					} catch (Exception ex) {
						System.out.println(ex);
					}
				}
			});
		} catch(Exception ex) {
			System.out.println(ex.getMessage()+";"+ex);
		} finally {
			try {
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}
